{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "google",
   "metadata": {},
   "source": [
    "##### Copyright 2025 Google LLC."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "apache",
   "metadata": {},
   "source": [
    "Licensed under the Apache License, Version 2.0 (the \"License\");\n",
    "you may not use this file except in compliance with the License.\n",
    "You may obtain a copy of the License at\n",
    "\n",
    "    http://www.apache.org/licenses/LICENSE-2.0\n",
    "\n",
    "Unless required by applicable law or agreed to in writing, software\n",
    "distributed under the License is distributed on an \"AS IS\" BASIS,\n",
    "WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n",
    "See the License for the specific language governing permissions and\n",
    "limitations under the License.\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "basename",
   "metadata": {},
   "source": [
    "# minimal_jobshop_sat"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "link",
   "metadata": {},
   "source": [
    "<table align=\"left\">\n",
    "<td>\n",
    "<a href=\"https://colab.research.google.com/github/google/or-tools/blob/main/examples/notebook/sat/minimal_jobshop_sat.ipynb\"><img src=\"https://raw.githubusercontent.com/google/or-tools/main/tools/colab_32px.png\"/>Run in Google Colab</a>\n",
    "</td>\n",
    "<td>\n",
    "<a href=\"https://github.com/google/or-tools/blob/main/ortools/sat/samples/minimal_jobshop_sat.py\"><img src=\"https://raw.githubusercontent.com/google/or-tools/main/tools/github_32px.png\"/>View source on GitHub</a>\n",
    "</td>\n",
    "</table>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "doc",
   "metadata": {},
   "source": [
    "First, you must install [ortools](https://pypi.org/project/ortools/) package in this colab."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "install",
   "metadata": {},
   "outputs": [],
   "source": [
    "%pip install ortools"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "description",
   "metadata": {},
   "source": [
    "\n",
    "Minimal jobshop example."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "code",
   "metadata": {},
   "outputs": [],
   "source": [
    "import collections\n",
    "from ortools.sat.python import cp_model\n",
    "\n",
    "\n",
    "\n",
    "def main() -> None:\n",
    "    \"\"\"Minimal jobshop problem.\"\"\"\n",
    "    # Data.\n",
    "    jobs_data = [  # task = (machine_id, processing_time).\n",
    "        [(0, 3), (1, 2), (2, 2)],  # Job0\n",
    "        [(0, 2), (2, 1), (1, 4)],  # Job1\n",
    "        [(1, 4), (2, 3)],  # Job2\n",
    "    ]\n",
    "\n",
    "    machines_count = 1 + max(task[0] for job in jobs_data for task in job)\n",
    "    all_machines = range(machines_count)\n",
    "    # Computes horizon dynamically as the sum of all durations.\n",
    "    horizon = sum(task[1] for job in jobs_data for task in job)\n",
    "\n",
    "    # Create the model.\n",
    "    model = cp_model.CpModel()\n",
    "\n",
    "    # Named tuple to store information about created variables.\n",
    "    task_type = collections.namedtuple(\"task_type\", \"start end interval\")\n",
    "    # Named tuple to manipulate solution information.\n",
    "    assigned_task_type = collections.namedtuple(\n",
    "        \"assigned_task_type\", \"start job index duration\"\n",
    "    )\n",
    "\n",
    "    # Creates job intervals and add to the corresponding machine lists.\n",
    "    all_tasks = {}\n",
    "    machine_to_intervals = collections.defaultdict(list)\n",
    "\n",
    "    for job_id, job in enumerate(jobs_data):\n",
    "        for task_id, task in enumerate(job):\n",
    "            machine, duration = task\n",
    "            suffix = f\"_{job_id}_{task_id}\"\n",
    "            start_var = model.new_int_var(0, horizon, \"start\" + suffix)\n",
    "            end_var = model.new_int_var(0, horizon, \"end\" + suffix)\n",
    "            interval_var = model.new_interval_var(\n",
    "                start_var, duration, end_var, \"interval\" + suffix\n",
    "            )\n",
    "            all_tasks[job_id, task_id] = task_type(\n",
    "                start=start_var, end=end_var, interval=interval_var\n",
    "            )\n",
    "            machine_to_intervals[machine].append(interval_var)\n",
    "\n",
    "    # Create and add disjunctive constraints.\n",
    "    for machine in all_machines:\n",
    "        model.add_no_overlap(machine_to_intervals[machine])\n",
    "\n",
    "    # Precedences inside a job.\n",
    "    for job_id, job in enumerate(jobs_data):\n",
    "        for task_id in range(len(job) - 1):\n",
    "            model.add(\n",
    "                all_tasks[job_id, task_id + 1].start >= all_tasks[job_id, task_id].end\n",
    "            )\n",
    "\n",
    "    # Makespan objective.\n",
    "    obj_var = model.new_int_var(0, horizon, \"makespan\")\n",
    "    model.add_max_equality(\n",
    "        obj_var,\n",
    "        [all_tasks[job_id, len(job) - 1].end for job_id, job in enumerate(jobs_data)],\n",
    "    )\n",
    "    model.minimize(obj_var)\n",
    "\n",
    "    # Creates the solver and solve.\n",
    "    solver = cp_model.CpSolver()\n",
    "    status = solver.solve(model)\n",
    "\n",
    "    if status == cp_model.OPTIMAL or status == cp_model.FEASIBLE:\n",
    "        print(\"Solution:\")\n",
    "        # Create one list of assigned tasks per machine.\n",
    "        assigned_jobs = collections.defaultdict(list)\n",
    "        for job_id, job in enumerate(jobs_data):\n",
    "            for task_id, task in enumerate(job):\n",
    "                machine = task[0]\n",
    "                assigned_jobs[machine].append(\n",
    "                    assigned_task_type(\n",
    "                        start=solver.value(all_tasks[job_id, task_id].start),\n",
    "                        job=job_id,\n",
    "                        index=task_id,\n",
    "                        duration=task[1],\n",
    "                    )\n",
    "                )\n",
    "\n",
    "        # Create per machine output lines.\n",
    "        output = \"\"\n",
    "        for machine in all_machines:\n",
    "            # Sort by starting time.\n",
    "            assigned_jobs[machine].sort()\n",
    "            sol_line_tasks = \"Machine \" + str(machine) + \": \"\n",
    "            sol_line = \"           \"\n",
    "\n",
    "            for assigned_task in assigned_jobs[machine]:\n",
    "                name = f\"job_{assigned_task.job}_task_{assigned_task.index}\"\n",
    "                # add spaces to output to align columns.\n",
    "                sol_line_tasks += f\"{name:15}\"\n",
    "\n",
    "                start = assigned_task.start\n",
    "                duration = assigned_task.duration\n",
    "                sol_tmp = f\"[{start},{start + duration}]\"\n",
    "                # add spaces to output to align columns.\n",
    "                sol_line += f\"{sol_tmp:15}\"\n",
    "\n",
    "            sol_line += \"\\n\"\n",
    "            sol_line_tasks += \"\\n\"\n",
    "            output += sol_line_tasks\n",
    "            output += sol_line\n",
    "\n",
    "        # Finally print the solution found.\n",
    "        print(f\"Optimal Schedule Length: {solver.objective_value}\")\n",
    "        print(output)\n",
    "    else:\n",
    "        print(\"No solution found.\")\n",
    "\n",
    "    # Statistics.\n",
    "    print(\"\\nStatistics\")\n",
    "    print(f\"  - conflicts: {solver.num_conflicts}\")\n",
    "    print(f\"  - branches : {solver.num_branches}\")\n",
    "    print(f\"  - wall time: {solver.wall_time}s\")\n",
    "\n",
    "\n",
    "main()\n",
    "\n"
   ]
  }
 ],
 "metadata": {
  "language_info": {
   "name": "python"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}
